fix(agentex): end SSE subscriptions on terminal task / vanished topic; stop deleting shared stream by deepthi-rao-scale · Pull Request #385 · scaleapi/scale-agentex

deepthi-rao-scale · 2026-07-29T16:25:17Z

What was broken

When you open a task in the UI, the backend keeps a live stream open to push updates. That stream had no way to end itself — it only stopped if the browser disconnected.

So when a task finished, the backend just kept listening forever, holding one Redis connection each time. These pile up until the pool is empty, at which point the pod can't serve requests (and its health check fails, so it sits stuck until a manual restart).

A second bug: when one viewer left, the code deleted the task's shared event stream — yanking it out from under everyone else watching the same task.

The fix

End the stream when the task is done. The backend already gets a task_updated event when a task finishes, so the stream now closes as soon as it sees a terminal status.
Fallback for the rare cases the event can't cover (viewer connects after the task already finished, or the producer dies silently): a lightweight check on the existing keepalive interval closes the stream then too.
Stop deleting the shared stream on one viewer's exit — the Redis TTL already cleans it up on its own.

Testing

Two integration regression tests (stream self-ends on terminal task; one viewer disconnecting doesn't delete the shared stream).

Verified live (A/B against main) — same steps both times: create a RUNNING task → open GET /tasks/{id}/stream → complete the task, while watching Redis blocked_clients.

	on task completion	Redis connection
`main`	stream keeps `XREAD`-looping; kept sending `:ping` 15s+ after completion and only ended when the client was force-disconnected	`blocked_clients` stayed `1` — released only on client disconnect (zombie)
this branch	stream delivers the terminal `task_updated` and returns immediately (`Ending SSE stream …: received a terminal task_updated event`)	`blocked_clients` `1 → 0` instantly, no client disconnect needed

On main the reader logged Reading messages from Redis stream … every ~2s indefinitely after the task was done; on this branch it stops the moment the terminal event arrives. Same reproduction the incident described, and the connection is released instead of pinned.

Note: the unit/integration suite runs in CI; the tutorial-agent integration jobs were red due to a pre-existing missing-pytest-asyncio issue unrelated to this change (fixed separately in scale-agentex-python).

Related: the UI-side half of this leak is #383.

🤖 Generated with Claude Code

Greptile Summary

This PR changes task SSE subscription lifecycle management.

Ends subscriptions after terminal task updates or a keepalive-cadence fallback check.
Leaves shared Redis streams intact when individual subscribers disconnect.
Adds Redis stream-existence support and integration regression coverage.

Confidence Score: 4/5

The PR is not yet safe to merge because live subscriptions can still close on recoverable FAILED states or when a transient task lookup failure coincides with an expired stream key.

FAILED remains an exit condition despite an existing FAILED-to-RUNNING transition, and the fallback treats all database lookup failures as an absent task, allowing an inconclusive lookup to terminate a live subscription before later events recreate its stream.

Files Needing Attention: agentex/src/domain/use_cases/streams_use_case.py

Important Files Changed

Filename	Overview
agentex/src/domain/use_cases/streams_use_case.py	Adds event-driven and fallback subscription termination, but still closes on recoverable FAILED states and can mistake transient lookup failures for vanished tasks.
agentex/src/adapters/streams/adapter_redis.py	Adds an asynchronous Redis EXISTS wrapper for stream lifecycle checks.
agentex/src/adapters/streams/port.py	Extends the stream repository contract with stream_exists.
agentex/tests/integration/test_task_stream.py	Adds regression coverage for terminal closure, shared-stream retention, late terminal subscriptions, and reclaimed keys with successful task lookups.

Sequence Diagram

sequenceDiagram
  participant Client
  participant Streams as StreamsUseCase
  participant Tasks as TaskService
  participant Redis
  Client->>Streams: Subscribe to task events
  loop Read events
    Streams->>Redis: XREAD stream
    Redis-->>Streams: task_updated / timeout
    alt Terminal task_updated
      Streams-->>Client: Deliver terminal event
      Streams-->>Client: End SSE
    else Keepalive interval
      Streams->>Tasks: get_task(task_id)
      Streams->>Redis: EXISTS stream
      alt Task terminal or definitively vanished
        Streams-->>Client: End SSE
      else Task still live
        Streams-->>Client: :ping
      end
    end
  end

Prompt To Fix All With AI

### Issue 1
agentex/src/domain/use_cases/streams_use_case.py:256-263
**Transient lookup closes live stream**

When a transient database lookup failure occurs after a live nonterminal task's Redis stream key has expired, this handler converts the failure to `task=None` and treats the absent topic as finished. The SSE generator then exits, causing the viewer to miss subsequent events when a later `XADD` recreates the stream.

---

For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

_{Reviews (3): Last reviewed commit: "test(agentex): cover SSE fallback termin..." | Re-trigger Greptile}

Greptile also left 1 inline comment on this PR.

…hared stream Task event SSE subscriptions had no terminal condition. stream_task_events ran a `while True` loop whose only exits were client disconnect (CancelledError) and fatal errors. Once a task finished its producer stopped writing, but the reader kept blocking on the topic forever — issuing an XREAD every couple of seconds and pinning one connection from the shared per-process Redis pool. Because the stream key carries a sliding TTL, a finished task's key eventually disappears while readers keep blocking on it, turning each subscription into a permanent zombie. Accumulated zombies exhaust the pool, which is shared with the readiness probe, so /readyz fails while the dependency-free /healthz keeps returning 200 — the pod goes Unready and is never restarted. Termination: - Primary is event-driven. Every terminal transition funnels through task_service.transition_status (and delete via update_task), which XADDs a task_updated event carrying the new status onto the task's own stream. The reader already delivers those events, so it returns as soon as it forwards one whose task is in a terminal status — no DB lookup and no ordering race, since the terminal event is the last message produced. - Fallback runs on the keepalive-ping cadence for the cases the event path cannot see: a client that connects after the task already finished, a producer that dies without emitting a terminal event, or a task_updated with no task payload. It ends the stream when the task is terminal or when a topic that had existed is reclaimed. stream_ever_existed is seeded from the tail snapshot so a mid-flight subscriber correctly arms the vanished-topic check. Shared stream: - Drop the per-subscriber cleanup_stream in finally. The topic is keyed only by task id and shared by every viewer, so deleting it on one subscriber's exit tore the stream out from under the others. The sliding TTL already reclaims it. Also adds stream_exists to the stream port + Redis adapter (EXISTS) for the vanished-topic fallback, and two integration regression tests: the stream ends on its own once the task is terminal, and a single subscriber disconnecting does not delete the shared topic. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

rpatel-scale · 2026-07-30T20:31:33Z

+                            if await self._stream_is_finished(
+                                task_id, stream_topic, stream_ever_existed
+                            ):


Why do we need here and above? Wouldnt above already have executed by now?

Address review: the fallback vanished-topic check closed the SSE stream for a still-running task whose sliding-TTL stream key was reclaimed after an idle period, even though a later XADD would recreate the key. Only treat a vanished topic as terminal when the task's status could not be confirmed (lookup failed / hard delete); a confirmed non-terminal task stays live. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Add two integration regression tests that had no coverage: - test_stream_ends_on_late_connect_to_terminal_task: connecting after a task is already terminal never delivers a terminal event, so the keepalive-cadence fallback must end the stream. - test_running_task_stream_survives_reclaimed_key: a still-RUNNING task whose idle stream key is reclaimed by the sliding TTL must NOT be closed (guards the review fix in the prior commit). Both shrink SSE_KEEPALIVE_PING_INTERVAL so the fallback runs in ~1s. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

greptile-apps · 2026-07-30T21:06:42Z

+        try:
+            task = await self.task_service.get_task(id=task_id)
+        except Exception as e:
+            logger.warning(
+                f"Terminal-state lookup failed for task {task_id}; "
+                f"treating status as unknown: {e}"
+            )
+            task = None


Transient lookup closes live stream

When a transient database lookup failure occurs after a live nonterminal task's Redis stream key has expired, this handler converts the failure to task=None and treats the absent topic as finished. The SSE generator then exits, causing the viewer to miss subsequent events when a later XADD recreates the stream.

Prompt To Fix With AI

This is a comment left during a code review. Path: agentex/src/domain/use_cases/streams_use_case.py Line: 256-263 Comment: **Transient lookup closes live stream** When a transient database lookup failure occurs after a live nonterminal task's Redis stream key has expired, this handler converts the failure to `task=None` and treats the absent topic as finished. The SSE generator then exits, causing the viewer to miss subsequent events when a later `XADD` recreates the stream. --- For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.

deepthi-rao-scale force-pushed the fix/agentex-sse-zombie-subscription-leak branch from 89b915c to 2846998 Compare July 29, 2026 18:16

deepthi-rao-scale force-pushed the fix/agentex-sse-zombie-subscription-leak branch from 2846998 to 2eed3af Compare July 29, 2026 18:35

deepthi-rao-scale and others added 2 commits July 30, 2026 12:23

docs: tighten the zombie-subscription comment in stream_task_events

d9d2fa6

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Merge branch 'main' into fix/agentex-sse-zombie-subscription-leak

9ec9442

deepthi-rao-scale marked this pull request as ready for review July 30, 2026 20:23

deepthi-rao-scale requested a review from a team as a code owner July 30, 2026 20:23

greptile-apps Bot reviewed Jul 30, 2026

View reviewed changes

Comment thread agentex/src/domain/use_cases/streams_use_case.py

Comment thread agentex/src/domain/use_cases/streams_use_case.py Outdated

rpatel-scale approved these changes Jul 30, 2026

View reviewed changes

rpatel-scale reviewed Jul 30, 2026

View reviewed changes

deepthi-rao-scale and others added 2 commits July 30, 2026 16:46

greptile-apps Bot reviewed Jul 30, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

fix(agentex): end SSE subscriptions on terminal task / vanished topic; stop deleting shared stream - #385

fix(agentex): end SSE subscriptions on terminal task / vanished topic; stop deleting shared stream#385
deepthi-rao-scale wants to merge 5 commits into
mainfrom
fix/agentex-sse-zombie-subscription-leak

deepthi-rao-scale commented Jul 29, 2026 •

edited by greptile-apps Bot

Loading

Uh oh!

Uh oh!

Uh oh!

rpatel-scale Jul 30, 2026

Uh oh!

greptile-apps Bot Jul 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

deepthi-rao-scale commented Jul 29, 2026 • edited by greptile-apps Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What was broken

The fix

Testing

Greptile Summary

Confidence Score: 4/5

Important Files Changed

Sequence Diagram

Uh oh!

Uh oh!

Uh oh!

rpatel-scale Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

deepthi-rao-scale commented Jul 29, 2026 •

edited by greptile-apps Bot

Loading